/**
 * @file reqrep_server.c
 * @brief nanomsg REQ/REP 模式服务器端演示
 * 
 * 这个示例展示了如何使用 nanomsg 的 REP (Reply) 模式实现服务器。
 * REQ/REP 模式提供了请求-响应通信模型，支持负载均衡和故障转移。
 */

#include "common.h"

// 全局变量
volatile int g_running = 1;
stats_t g_stats = {0};

/**
 * @brief 处理客户端请求
 * @param request 请求消息
 * @param response 响应缓冲区
 * @param response_size 响应缓冲区大小
 * @return 响应数据长度
 */
int process_request(const message_t* request, char* response, size_t response_size) {
    if (!request || !response) {
        return -1;
    }
    
    // 解析请求类型并生成相应的响应
    switch (request->type) {
        case MSG_TEXT: {
            // 简单的文本回复
            snprintf(response, response_size, "Echo: %s [seq:%u, time:%u]", 
                    request->data, request->sequence, request->timestamp);
            break;
        }
        
        case MSG_JSON: {
            // JSON格式回复
            snprintf(response, response_size, 
                    "{\"status\":\"ok\",\"echo\":\"%s\",\"seq\":%u,\"timestamp\":%u}", 
                    request->data, request->sequence, get_timestamp());
            break;
        }
        
        case MSG_HEARTBEAT: {
            // 心跳响应
            snprintf(response, response_size, "HEARTBEAT_ACK");
            break;
        }
        
        default: {
            // 未知请求类型
            snprintf(response, response_size, "ERROR: Unknown request type %d", request->type);
            break;
        }
    }
    
    return strlen(response);
}

/**
 * @brief 运行REP服务器
 * @param address 绑定地址
 * @return 成功返回0，失败返回-1
 */
int run_rep_server(const char* address) {
    int sock;
    char recv_buffer[MAX_MESSAGE_SIZE];
    char send_buffer[MAX_MESSAGE_SIZE];
    char response_data[BUFFER_SIZE];
    
    log_message(LOG_INFO, "Starting REP server on %s", address);
    
    // 创建 REP 套接字
    sock = nn_socket(AF_SP, NN_REP);
    CHECK_SOCKET(sock);
    
    // 设置套接字选项
    int timeout = 5000; // 5秒超时
    if (nn_setsockopt(sock, NN_SOL_SOCKET, NN_RCVTIMEO, &timeout, sizeof(timeout)) < 0) {
        log_message(LOG_WARN, "Failed to set receive timeout: %s", nn_strerror(nn_errno()));
    }
    
    // 绑定地址
    int bind_result = nn_bind(sock, address);
    CHECK_BIND(bind_result, address);
    
    log_message(LOG_INFO, "REP server listening on %s", address);
    log_message(LOG_INFO, "Waiting for client requests...");
    
    while (should_continue()) {
        // 接收请求
        int bytes_received = nn_recv(sock, recv_buffer, sizeof(recv_buffer), 0);
        
        if (bytes_received > 0) {
            update_recv_stats(bytes_received);
            
            // 反序列化请求消息
            message_t* request = deserialize_message(recv_buffer, bytes_received);
            if (request) {
                log_message(LOG_INFO, "Received request [seq:%u]: %s", 
                           request->sequence, request->data);
                
                // 处理请求
                int response_len = process_request(request, response_data, sizeof(response_data));
                
                if (response_len > 0) {
                    // 创建响应消息
                    message_t* response = create_message(request->type, response_data, response_len);
                    if (response) {
                        // 序列化响应消息
                        int serialized_len = serialize_message(response, send_buffer, sizeof(send_buffer));
                        
                        if (serialized_len > 0) {
                            // 发送响应
                            int bytes_sent = nn_send(sock, send_buffer, serialized_len, 0);
                            if (bytes_sent > 0) {
                                update_send_stats(bytes_sent);
                                log_message(LOG_INFO, "Sent response [%d bytes]: %s", 
                                           bytes_sent, response_data);
                            } else {
                                log_message(LOG_ERROR, "Failed to send response: %s", 
                                           nn_strerror(nn_errno()));
                                update_error_stats();
                            }
                        } else {
                            log_message(LOG_ERROR, "Failed to serialize response");
                            update_error_stats();
                        }
                        
                        destroy_message(response);
                    }
                } else {
                    log_message(LOG_ERROR, "Failed to process request");
                    update_error_stats();
                }
                
                destroy_message(request);
            } else {
                log_message(LOG_ERROR, "Failed to deserialize request");
                update_error_stats();
            }
        } else if (bytes_received < 0) {
            int err = nn_errno();
            if (err == ETERM || err == EINTR) {
                break; // 正常退出
            } else if (err == ETIMEDOUT || err == EAGAIN) {
                // 超时，继续等待
                continue;
            } else {
                log_message(LOG_ERROR, "Receive error: %s", nn_strerror(err));
                update_error_stats();
            }
        }
    }
    
    nn_close(sock);
    log_message(LOG_INFO, "REP server stopped");
    return 0;
}

/**
 * @brief 运行多线程REP服务器（演示负载均衡）
 * @param address 绑定地址
 * @param worker_count 工作线程数量
 * @return 成功返回0，失败返回-1
 */
int run_multithreaded_server(const char* address, int worker_count) {
    log_message(LOG_INFO, "Multi-threaded server not implemented in this demo");
    log_message(LOG_INFO, "For production use, consider using thread pools or process pools");
    
    // 这里可以实现多线程版本，使用 pthread 创建多个工作线程
    // 每个线程运行独立的 REP 套接字
    
    return run_rep_server(address);
}

/**
 * @brief 打印使用说明
 */
void print_server_usage(const char* program_name) {
    printf("Usage: %s [options]\n", program_name);
    printf("\n");
    printf("Options:\n");
    printf("  -h, --help         Show this help message\n");
    printf("  -v, --version      Show version information\n");
    printf("  -a, --address ADDR Bind address (default: %s)\n", DEFAULT_TCP_ADDR);
    printf("  -w, --workers NUM  Number of worker threads (default: 1)\n");
    printf("  -t, --timeout MS   Receive timeout in milliseconds (default: 5000)\n");
    printf("  -d, --daemon       Run as daemon (background)\n");
    printf("\n");
    printf("Supported address formats:\n");
    printf("  tcp://*:port      - TCP server on all interfaces\n");
    printf("  tcp://ip:port     - TCP server on specific interface\n");
    printf("  ipc://path        - IPC server\n");
    printf("  inproc://name     - In-process server\n");
    printf("\n");
    printf("Examples:\n");
    printf("  %s                              # Start server on default address\n", program_name);
    printf("  %s -a tcp://*:8080              # Start on TCP port 8080\n", program_name);
    printf("  %s -a ipc:///tmp/myserver.ipc   # Start on IPC socket\n", program_name);
    printf("  %s -w 4                         # Start with 4 worker threads\n", program_name);
}

/**
 * @brief 主函数
 */
int main(int argc, char* argv[]) {
    const char* address = DEFAULT_TCP_ADDR;
    int worker_count = 1;
    int timeout = 5000;
    int daemon_mode = 0;
    
    // 解析命令行参数
    for (int i = 1; i < argc; i++) {
        if (strcmp(argv[i], "-h") == 0 || strcmp(argv[i], "--help") == 0) {
            print_server_usage(argv[0]);
            return 0;
        } else if (strcmp(argv[i], "-v") == 0 || strcmp(argv[i], "--version") == 0) {
            print_nanomsg_info();
            return 0;
        } else if (strcmp(argv[i], "-a") == 0 || strcmp(argv[i], "--address") == 0) {
            if (i + 1 < argc) {
                address = argv[++i];
            } else {
                log_message(LOG_ERROR, "Option %s requires an argument", argv[i]);
                return 1;
            }
        } else if (strcmp(argv[i], "-w") == 0 || strcmp(argv[i], "--workers") == 0) {
            if (i + 1 < argc) {
                worker_count = atoi(argv[++i]);
                if (worker_count < 1 || worker_count > 100) {
                    log_message(LOG_ERROR, "Invalid worker count: %d", worker_count);
                    return 1;
                }
            } else {
                log_message(LOG_ERROR, "Option %s requires an argument", argv[i]);
                return 1;
            }
        } else if (strcmp(argv[i], "-t") == 0 || strcmp(argv[i], "--timeout") == 0) {
            if (i + 1 < argc) {
                timeout = atoi(argv[++i]);
                if (timeout < 100 || timeout > 60000) {
                    log_message(LOG_ERROR, "Invalid timeout: %d ms", timeout);
                    return 1;
                }
            } else {
                log_message(LOG_ERROR, "Option %s requires an argument", argv[i]);
                return 1;
            }
        } else if (strcmp(argv[i], "-d") == 0 || strcmp(argv[i], "--daemon") == 0) {
            daemon_mode = 1;
        } else {
            log_message(LOG_ERROR, "Unknown option: %s", argv[i]);
            print_server_usage(argv[0]);
            return 1;
        }
    }
    
    // 初始化
    setup_signal_handlers();
    init_stats();
    check_nanomsg_version();
    
    // 守护进程模式
    if (daemon_mode) {
        if (daemon(0, 0) != 0) {
            log_message(LOG_ERROR, "Failed to create daemon: %s", strerror(errno));
            return 1;
        }
        log_message(LOG_INFO, "Running in daemon mode");
    }
    
    log_message(LOG_INFO, "=== Nanomsg REQ/REP Server Demo ===");
    log_message(LOG_INFO, "Address: %s (%s)", address, parse_address_type(address));
    log_message(LOG_INFO, "Workers: %d", worker_count);
    log_message(LOG_INFO, "Timeout: %d ms", timeout);
    log_message(LOG_INFO, "Press Ctrl+C to stop");
    
    int result;
    if (worker_count > 1) {
        result = run_multithreaded_server(address, worker_count);
    } else {
        result = run_rep_server(address);
    }
    
    // 打印统计信息
    log_message(LOG_INFO, "Server shutting down...");
    print_stats();
    
    return result;
}

// 实现必要的公共函数（简化版）
void init_stats(void) {
    memset(&g_stats, 0, sizeof(g_stats));
    g_stats.start_time = time(NULL);
}

void print_stats(void) {
    time_t duration = time(NULL) - g_stats.start_time;
    printf("\n" COLOR_CYAN "=== Server Statistics ===" COLOR_RESET "\n");
    printf("Uptime: %ld seconds\n", duration);
    printf("Requests handled: %lu\n", g_stats.messages_received);
    printf("Responses sent: %lu\n", g_stats.messages_sent);
    printf("Bytes received: %lu\n", g_stats.bytes_received);
    printf("Bytes sent: %lu\n", g_stats.bytes_sent);
    printf("Errors: %u\n", g_stats.errors);
    if (duration > 0) {
        printf("Request rate: %.2f req/s\n", (double)g_stats.messages_received / duration);
        printf("Response rate: %.2f resp/s\n", (double)g_stats.messages_sent / duration);
        printf("Throughput: %.2f KB/s (in), %.2f KB/s (out)\n", 
               (double)g_stats.bytes_received / 1024 / duration,
               (double)g_stats.bytes_sent / 1024 / duration);
    }
}

void log_message(log_level_t level, const char* format, ...) {
    const char* level_colors[] = {COLOR_BLUE, COLOR_GREEN, COLOR_YELLOW, COLOR_RED};
    const char* level_names[] = {"DEBUG", "INFO", "WARN", "ERROR"};
    
    char timestamp[32];
    time_t now = time(NULL);
    struct tm* tm_info = localtime(&now);
    strftime(timestamp, sizeof(timestamp), "%H:%M:%S", tm_info);
    
    va_list args;
    va_start(args, format);
    
    printf("%s[%s] %s%s" COLOR_RESET " ", 
           level_colors[level], timestamp, level_names[level], COLOR_RESET);
    vprintf(format, args);
    printf("\n");
    
    va_end(args);
    fflush(stdout);
}

void handle_error(const char* message) {
    log_message(LOG_ERROR, "%s: %s", message, nn_strerror(nn_errno()));
    update_error_stats();
}

void signal_handler(int sig) {
    if (sig == SIGINT || sig == SIGTERM) {
        log_message(LOG_INFO, "Received signal %d, shutting down gracefully...", sig);
        g_running = 0;
    }
}

void setup_signal_handlers(void) {
    signal(SIGINT, signal_handler);
    signal(SIGTERM, signal_handler);
    signal(SIGPIPE, SIG_IGN); // 忽略管道破裂信号
}

message_t* create_message(message_type_t type, const char* data, uint32_t length) {
    static uint32_t sequence = 0;
    message_t* msg = malloc(sizeof(message_t));
    if (!msg) return NULL;
    
    msg->type = type;
    msg->sequence = ++sequence;
    msg->timestamp = get_timestamp();
    msg->data_length = length;
    
    if (length > 0 && data) {
        memcpy(msg->data, data, length);
    }
    
    return msg;
}

void destroy_message(message_t* msg) {
    if (msg) free(msg);
}

int serialize_message(const message_t* msg, char* buffer, size_t buffer_size) {
    if (!msg || !buffer) return -1;
    
    size_t total_size = sizeof(message_type_t) + sizeof(uint32_t) * 3 + msg->data_length;
    if (total_size > buffer_size) return -1;
    
    char* ptr = buffer;
    memcpy(ptr, &msg->type, sizeof(msg->type)); ptr += sizeof(msg->type);
    memcpy(ptr, &msg->sequence, sizeof(msg->sequence)); ptr += sizeof(msg->sequence);
    memcpy(ptr, &msg->timestamp, sizeof(msg->timestamp)); ptr += sizeof(msg->timestamp);
    memcpy(ptr, &msg->data_length, sizeof(msg->data_length)); ptr += sizeof(msg->data_length);
    memcpy(ptr, msg->data, msg->data_length);
    
    return total_size;
}

message_t* deserialize_message(const char* buffer, size_t length) {
    if (!buffer || length < sizeof(message_type_t) + sizeof(uint32_t) * 3) return NULL;
    
    message_t* msg = malloc(sizeof(message_t));
    if (!msg) return NULL;
    
    const char* ptr = buffer;
    memcpy(&msg->type, ptr, sizeof(msg->type)); ptr += sizeof(msg->type);
    memcpy(&msg->sequence, ptr, sizeof(msg->sequence)); ptr += sizeof(msg->sequence);
    memcpy(&msg->timestamp, ptr, sizeof(msg->timestamp)); ptr += sizeof(msg->timestamp);
    memcpy(&msg->data_length, ptr, sizeof(msg->data_length)); ptr += sizeof(msg->data_length);
    
    if (msg->data_length > MAX_MESSAGE_SIZE || 
        length < sizeof(message_type_t) + sizeof(uint32_t) * 3 + msg->data_length) {
        free(msg);
        return NULL;
    }
    
    memcpy(msg->data, ptr, msg->data_length);
    return msg;
}

uint32_t get_timestamp(void) {
    return (uint32_t)time(NULL);
}

const char* parse_address_type(const char* addr) {
    if (strncmp(addr, "tcp://", 6) == 0) return "TCP";
    if (strncmp(addr, "ipc://", 6) == 0) return "IPC";
    if (strncmp(addr, "inproc://", 9) == 0) return "INPROC";
    return "UNKNOWN";
}

void check_nanomsg_version(void) {
    log_message(LOG_INFO, "Nanomsg REP server initialized");
}

void print_nanomsg_info(void) {
    printf("Nanomsg REQ/REP Server Demo v1.0\n");
    printf("Built with nanomsg library\n");
    printf("Supports TCP, IPC, and INPROC transports\n");
    printf("Features: Load balancing, fault tolerance\n");
}